package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Demo6REpartition {
  def main(args: Array[String]): Unit = {
    //创建spark sql环境（新版spark统一的入口）
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[8]")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    //获取sparkContext
    val sc: SparkContext = spark.sparkContext

    /**
     * spark读取文件的分区数
     * 1、默认一个block块对应一个分区
     * 2、可以通过在读文件时指定最小分区数，控制分区数
     */
    //读取数据
    val wordsRDD: RDD[String] = sc.textFile("data/words", 10)

    println(s"wordsRDD分区数：${wordsRDD.getNumPartitions}")

    /**
     * repartition： 重分区，会产生shuffle
     * 可以用于增加分区和减少分区
     */

    val reRDD: RDD[String] = wordsRDD.repartition(20)

    println(s"reRDD分区数：${reRDD.getNumPartitions}")

    /**
     * coalesce: 重分区，没有shuffle
     * 只能用于减少分区，不能用于增加分区
     */

    val cosRDD: RDD[String] = reRDD.coalesce(30)
    println(s"cosRDD分区数：${cosRDD.getNumPartitions}")


    /**
     * 在使用shuffle类算子时可以手动指定reduce端的分区数
     *
     */

    val countRDD: RDD[(String, Int)] = cosRDD
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey((x, y) => x + y, 2) //指定分区数

    println(s"countRDD分区数：${countRDD.getNumPartitions}")
  }

}
